package org.hh.jga.concurrent;

import java.util.Iterator;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Main Parallel static methods
 * 
 * @author Hong Hong
 *
 */
public final class Parallel {
	protected final static ThreadPoolExecutor s_executer = 
		new ThreadPoolExecutor(1, 8, 1L, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(16), new ExecutionHandler()); 
	protected static ParallelMode s_state = ParallelMode.MultiThread;
	protected static class ExecutionHandler implements RejectedExecutionHandler {
		public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
			r.run();
		}
	}
	public static enum ParallelMode {
		SingleThread,
		MultiThread
	}
	public static void setParallelMode(ParallelMode mode) {
		s_state = mode;
	}
	public static void resetThreadNumber(int maxNum) {
		s_executer.setMaximumPoolSize(maxNum);
	}
	
	public static <T> void foreach(Iterable<T> args, LoopBody<T> body) {
		if(args == null || body == null)
			return;
		doForEach(args, body, null, 1024);
	}
	public static <T> void foreach(Iterable<T> args, LoopBody<T> body, int chunkSize) {
		if(args == null || body == null)
			return;
		doForEach(args, body, null, chunkSize);
	}
	public static <T> void foreach(Iterable<T> args, LoopBody<T> body, LoopExceptionHandler<T> handler, int chunkSize) {
		if(args == null || body == null)
			return;
		doForEach(args, body, handler, chunkSize);
	}
	
	protected static <T> void doForEach(Iterable<T> args, LoopBody<T> body, LoopExceptionHandler<T> handler, int chunkSize) {
		if(s_state == ParallelMode.SingleThread) {
			for(T item : args) {
				try {
					body.body(item);
				} catch (Exception e) {
					if(handler != null)
						handler.handle(item, e);
				}
			}
		} else {
			Iterator<T> itr = args.iterator();
			Semaphore waiter = new Semaphore(0);
			int counter = 0;
			while(itr.hasNext()) {
				s_executer.execute(new LoopChunk<T>(waiter, body, handler, chunkSize, itr));
				++counter;
			}
			try {
				waiter.acquire(counter);
			} catch (InterruptedException e) { }
		}
	}
}
